package com.test.cn.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.immutable

case class LrisVo(var Id: String, var SepalLengthCm: Double, var SepalWidthCm: Double, var PetalLengthCm: Double, var PetalWidthCm: Double, var Species: String)

object KMeansQuestion {

  def main(args: Array[String]): Unit = {
    //
    val conf: SparkConf = new SparkConf()
      .setAppName(this.getClass.getCanonicalName.init)
      .setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    val K = 3
    //读取文件
    val textRDDA: RDD[String] = sc.textFile("spark_scala_home_work/src/data/Iris.csv")
    val lrisRDD: RDD[LrisVo] = textRDDA.map { case str => {
      val strs: Array[String] = str.split(",")
      LrisVo("", strs(1).toDouble, strs(2).toDouble, strs(3).toDouble, strs(4).toDouble, "")
    }
    }

    //随机找到三个点
    var voes: Array[LrisVo] = lrisRDD.take(3)
    var newVoes: Array[LrisVo] = Array()

    def getInstance(v1: LrisVo, v2: LrisVo): Double = {
      Math.sqrt(Math.pow((v1.PetalLengthCm - v2.PetalLengthCm), 2)
        + Math.pow(v1.PetalWidthCm - v2.PetalWidthCm, 2)
        + Math.pow(v1.SepalLengthCm - v2.SepalLengthCm, 2)
        + Math.pow(v1.SepalWidthCm - v2.SepalWidthCm, 2))
    }

    def sortMethod(v1: LrisVo): String = {
      "" + v1.SepalLengthCm + "_" + v1.SepalWidthCm + "_" + v1.PetalWidthCm + "_" + v1.PetalLengthCm
    }

    //循环的条件 1. 次数  2，中心点几乎不在进行移动
    def breakWhile(voes: Array[LrisVo], newVoes: Array[LrisVo]): Boolean = {
//      val tuples: Array[(LrisVo, LrisVo)] = voes.sortBy(sortMethod(_)).zip(newVoes.sortBy(sortMethod(_)))
      val tuples: Array[(LrisVo, LrisVo)] = voes.zip(newVoes)
      var flag = false
      for (elem <- tuples) {
        if (getInstance(elem._1, elem._2) > 0.000000000001) {
          flag = true
        }
      }
      flag
    }

    var flag = true
    while (flag) {
      //遍历数据集 计算每个点与3个中心的距离，距离那个中心点最近就属于那个中心点
      val mapValue: RDD[(Int, (LrisVo, Int))] = lrisRDD.map { case vo: LrisVo => {
        val tuples: immutable.IndexedSeq[Double] = for (i <- 0 until voes.length) yield {
          getInstance(voes(i), vo)
        }
        (tuples.indexOf(tuples.min), (vo, 1))
      }
      }

      def lrisAdd(v1: LrisVo, v2: LrisVo): LrisVo = {
        LrisVo(v1.Id,
          v1.SepalLengthCm + v2.SepalLengthCm,
          v1.SepalWidthCm + v2.SepalWidthCm,
          v1.PetalLengthCm + v2.PetalLengthCm,
          v1.PetalWidthCm + v2.PetalWidthCm,
          v1.Species)
      }

      //    val tuples: Array[(Int, (LrisVo, Int))] = mapValue.collect()
      //    for (elem <- tuples) {
      //      println(elem)
      //    }

      //根据新的分类计算新的中心点
      val reduceValue: RDD[(Int, (LrisVo, Int))] = mapValue.reduceByKey {
        case ((v1: LrisVo, num: Int), (v2: LrisVo, num2: Int)) => {
          (lrisAdd(v1, v2), num + num2)
        }
      }
      val finalValue: RDD[(Int, LrisVo)] = reduceValue.mapValues {
        case (v1: LrisVo, num: Int) => {
          LrisVo(v1.Id,
            v1.SepalLengthCm / num,
            v1.SepalWidthCm / num,
            v1.PetalLengthCm / num,
            v1.PetalWidthCm / num,
            v1.Species)
        }
      }

      //使用新的中心点开始下次的循环
      newVoes = finalValue.collect().sortBy(_._1).map(_._2)

      println("***************")


      if (breakWhile(voes, newVoes)) {
        flag = true
        voes = newVoes
      } else {
        flag = false
      }

    }

    for (elem <- newVoes) {
      println(elem)
    }


    sc.stop()
  }


}
